package net.bwie.dt.job.Doris;


import net.bwie.dt.bean.LogBean;
import net.bwie.dt.funtion.dtBeanMapFuctionLog;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import utils.DorisUtil;
import utils.KafkaUtil;

public class LogToDoris {
    public static void main(String[] args) throws Exception {
        // env执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);
        // 消费Kafka
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "new_log_table");
//        kafkaStream.print();
        // 数据转换
        DataStream<String> resultStream = healde(kafkaStream);

        resultStream.print();
        // 存入Doris
        DorisUtil.saveToDoris(
                resultStream,
                "dt_realtime_ranking",
                "log_table"
        );
        // 触发执行
        env.execute("LogToDoris");
    }

    private static DataStream<String> healde(DataStream<String> stream) {
        // 转实体类
        SingleOutputStreamOperator<LogBean> beanStream = stream.map(new dtBeanMapFuctionLog());


        SingleOutputStreamOperator<String> process = beanStream.process(new ProcessFunction<LogBean, String>() {
            @Override
            public void processElement(LogBean value, Context ctx, Collector<String> out) throws Exception {
                int log_id = value.getLog_id();
                int user_id = value.getUser_id();
                int product_id = value.getProduct_id();
                int number = value.getNumber();
                String action_type = value.getAction_type();
                String page_type = value.getPage_type();
                String event_type = value.getEvent_type();
                String ts = value.getTs();
                String output = log_id + "," + user_id + "," + product_id + ","+ number + "," +
                        action_type + "," + page_type+ "," + event_type+ "," + ts;
                out.collect(output);
            }
        });
        return process;
    }
}
